package com.elinshaw.hbase;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.Filter;
import org.apache.hadoop.hbase.filter.FilterList;
import org.apache.hadoop.hbase.filter.SingleColumnValueFilter;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;

public class MultiFilter {

    public static void main(String[] args) throws IOException {
        Configuration conf = HBaseConfiguration.create();
        conf.set("hbase.zookeeper.quorum", "datanode1:2181,datanode1:2181,datanode1:2181");
        Connection conn = ConnectionFactory.createConnection(conf);
        Table table = conn.getTable(TableName.valueOf("t2"));
        Scan scan = new Scan();
        // 创建过滤器1，查询年龄小于等于30岁的所有数据
        Filter filter1 = new SingleColumnValueFilter(Bytes.toBytes("f1"),
                Bytes.toBytes("age"), CompareOperator.LESS_OR_EQUAL,
                Bytes.toBytes("30"));
        // 创建过滤器2，查询年龄大于等于18岁的所有数据
        Filter filter2 = new SingleColumnValueFilter(Bytes.toBytes("f1"),
                Bytes.toBytes("age"), CompareOperator.GREATER_OR_EQUAL,
                Bytes.toBytes("18"));
        // 创建过滤器集合对象
        FilterList filterList = new FilterList();
        filterList.addFilter(filter1);
        filterList.addFilter(filter2);
        // 设置过滤器
        scan.setFilter(filterList);
        // 执行查询，得到结果集
        ResultScanner rs = table.getScanner(scan);
        // 输出结果，每个res代表一行数据
        for (Result res : rs) {
            if (!res.toString().equals("keyvalues=NONE")) {
                // 循环输出每个单元格的数据
                for (Cell cell : res.rawCells()) {
                    // 取得当前单元格所属的行键
                    String rowkey = new String(CellUtil.cloneRow(cell));
                    // 取得当前单元格所属的列族名称
                    String family = new String(CellUtil.cloneFamily(cell));
                    // 取得当前单元格所属的列名称
                    String qualifier = new String(CellUtil.cloneQualifier(cell));
                    // 取得当前单元格的列值
                    String value = new String(CellUtil.cloneValue(cell));
                    // 输出结果
                    System.out.println("行键：" + rowkey + "—————列：" + family + ":"
                            + qualifier + "—————值:" + value);
                }

            }

        }
        rs.close();
    }


}
